package avicit.bdp.dcs.json.service.writer;

import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.iotdb.param.IotdbDataxParam;
import avicit.bdp.dcs.stream.dto.MsgTemplateDto;
import avicit.bdp.dcs.stream.service.MsgTemplateService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.StreamColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.IotdbDataType;
import avicit.bdp.dcs.tools.enums.MeasureParaTypeEnum;
import avicit.bdp.dcs.tools.enums.ProcessMode;
import avicit.bdp.dcs.utils.JdbcConstants;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class IotdbWriterBuilder {

  private static final Logger logger = LoggerFactory.getLogger(IotdbWriterBuilder.class);

  @Autowired private MsgTemplateService msgTemplateService;

  @Value("${iotdb.batchSize}")
  private Integer batchSize;

  private static final int IOTDB_JDBC_URL_MIN_LEN = 13;
  private static final String DEVICEID = "deviceId";
  private static final String TIMESERIES = "timeseries";
  private static final String TYPE = "type";
  private static final String NAME = "name";

  /** 构建半结构化采集IotdbWriter */
  public BaseDataxParam buildIotdbWriterBySemi(
      JobDTO jobDTO,
      TaskDTO taskDTO,
      List<ColumnInfo> columnInfoList,
      Map<String, String> columnMap,
      List<ColumnMappingDto> columnMappingDtoList,
      JobDatasource writerDatasource) {
    // 1、构建Writer
    IotdbDataxParam dataxParam = new IotdbDataxParam();
    String jdbcUrl = writerDatasource.getJdbcUrl();
    if (jdbcUrl.length() > IOTDB_JDBC_URL_MIN_LEN) {
      jdbcUrl = jdbcUrl.substring(IOTDB_JDBC_URL_MIN_LEN, jdbcUrl.length() - 1);
    }
    dataxParam.setBootstrapServers(jdbcUrl);
    dataxParam.setUsername(writerDatasource.getJdbcUsername());
    dataxParam.setPassword(writerDatasource.getEncryptJdbcPassword());
    dataxParam.setDeviceId(taskDTO.getDstTableName());
    StringBuilder paras = new StringBuilder();

    // 2、转换列类型
    if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()
        || jobDTO.getProcessMode() == ProcessMode.PRECHECK_ERROR.getCode()) {
      if (CollectionUtils.isNotEmpty(columnInfoList)) {
        for (ColumnInfo columnInfo : columnInfoList) {
          if (columnMap.get(columnInfo.getName()) == null) {
            continue;
          }

          paras.append(columnMap.get(columnInfo.getName()));
          String type = columnInfo.getType();
          IotdbDataType iotdbDataType = IotdbDataType.TEXT;

          if (type.equals(MeasureParaTypeEnum.BOOLEAN.name())) {
            iotdbDataType = IotdbDataType.BOOLEAN;
          } else if (type.equals(MeasureParaTypeEnum.INTEGER.name())) {
            iotdbDataType = IotdbDataType.INT32;
          } else if (type.equals(MeasureParaTypeEnum.LONG.name())) {
            iotdbDataType = IotdbDataType.INT64;
          } else if (type.equals(MeasureParaTypeEnum.FLOAT.name())) {
            iotdbDataType = IotdbDataType.FLOAT;
          } else if (type.equals(MeasureParaTypeEnum.DOUBLE.name())) {
            iotdbDataType = IotdbDataType.DOUBLE;
          } else if (type.equals(MeasureParaTypeEnum.STRING.name())) {
            iotdbDataType = IotdbDataType.TEXT;
          }

          paras.append(Constants.COLON).append(iotdbDataType.getCode());
          paras.append(Constants.COMMA);
        }
      }
    } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()
        || jobDTO.getProcessMode() == ProcessMode.CLEAR_DST_TABLE_DATA.getCode()) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        paras.append(columnMap.get(columnMappingDto.getSrcColumnName()));
        String type = columnMappingDto.getDstColumnType();
        IotdbDataType iotdbDataType = IotdbDataType.TEXT;

        if (type.equals(MeasureParaTypeEnum.BOOLEAN.name())) {
          iotdbDataType = IotdbDataType.BOOLEAN;
        } else if (type.equals(MeasureParaTypeEnum.INTEGER.name())) {
          iotdbDataType = IotdbDataType.INT32;
        } else if (type.equals(MeasureParaTypeEnum.LONG.name())) {
          iotdbDataType = IotdbDataType.INT64;
        } else if (type.equals(MeasureParaTypeEnum.FLOAT.name())) {
          iotdbDataType = IotdbDataType.FLOAT;
        } else if (type.equals(MeasureParaTypeEnum.DOUBLE.name())) {
          iotdbDataType = IotdbDataType.DOUBLE;
        } else if (type.equals(MeasureParaTypeEnum.STRING.name())) {
          iotdbDataType = IotdbDataType.TEXT;
        }

        paras.append(Constants.COLON).append(iotdbDataType.getCode());
        paras.append(Constants.COMMA);
      }
    }

    String str = paras.toString();
    if (str.length() > 1) {
      str = str.substring(0, str.length() - 1);
    }
    dataxParam.setParas(str);
    dataxParam.setBatchSize(batchSize == null ? 50 : batchSize);
    dataxParam.setSourceType(JdbcConstants.HDFS);
    return dataxParam;
  }

  /** 构建流式采集IotdbWriter */
  public BaseDataxParam buildIotdbWriterByStream(
      JobDTO jobDTO,
      List<StreamColumnMappingDto> streamColumnMappingDtoList,
      JobDatasource writerDatasource) {
    // 1、构建Writer
    IotdbDataxParam dataxParam = new IotdbDataxParam();
    String jdbcUrl = writerDatasource.getJdbcUrl();
    if (jdbcUrl.length() > IOTDB_JDBC_URL_MIN_LEN) {
      jdbcUrl = jdbcUrl.substring(IOTDB_JDBC_URL_MIN_LEN, jdbcUrl.length() - 1);
    }
    dataxParam.setBootstrapServers(jdbcUrl);
    dataxParam.setUsername(writerDatasource.getJdbcUsername());
    dataxParam.setPassword(writerDatasource.getEncryptJdbcPassword());

    // 2、获取模板和表的对应关系
    List<Map<String, Object>> kafka2IotList = new ArrayList<>();
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    for (TableMappingDto tableMappingDto : tableMappingDtoList) {
      String templateId = tableMappingDto.getSrc_table();
      // 获取设备id
      MsgTemplateDto msgTemplateDtoQuery = msgTemplateService.selectByPrimaryKey(templateId);

      Map<String, Object> map = new HashMap<>();
      map.put(DEVICEID, msgTemplateDtoQuery.getDeviceId());
      map.put(TIMESERIES, tableMappingDto.getDst_table());
      kafka2IotList.add(map);
    }
    dataxParam.setKafka2IotMap(kafka2IotList);

    // 3、设置writerColumn
    List<IotdbDataxParam.IotdbParam> writerColumns = new ArrayList<>();
    for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {
      IotdbDataxParam.IotdbParam dataxIotdbParasPojo = new IotdbDataxParam.IotdbParam();
      dataxIotdbParasPojo.setTimeseries(streamColumnMappingDto.getTableName());

      // 获取设备id
      MsgTemplateDto msgTemplateDtoQuery =
          msgTemplateService.selectByPrimaryKey(streamColumnMappingDto.getTemplateId());
      dataxIotdbParasPojo.setDeviceId(msgTemplateDtoQuery.getDeviceId());
      List<Map<String, Object>> list = new ArrayList<>();

      List<ColumnMappingDto> columnList = streamColumnMappingDto.getColumnList();
      for (ColumnMappingDto columnMappingDto : columnList) {
        Map<String, Object> map = new HashMap<>();
        map.put(NAME, columnMappingDto.getDstColumnName());
        if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()
            || jobDTO.getProcessMode() == ProcessMode.PRECHECK_ERROR.getCode()) {
          // 如果是新建表，需要根据模板和iotdb数据类型进行转换
          String type = columnMappingDto.getDstColumnType();
          if (type.equalsIgnoreCase(MeasureParaTypeEnum.STRING.getType())
              || type.equalsIgnoreCase(IotdbDataType.TEXT.getDescp())) {
            map.put(TYPE, IotdbDataType.TEXT.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.BOOLEAN.getType())) {
            map.put(TYPE, IotdbDataType.BOOLEAN.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.BYTE.getType())
              || type.equalsIgnoreCase(IotdbDataType.INT32.getDescp())) {
            map.put(TYPE, IotdbDataType.INT32.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.CHAR.getType())) {
            map.put(TYPE, IotdbDataType.TEXT.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.DATE.getType())) {
            map.put(TYPE, IotdbDataType.TEXT.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.DOUBLE.getType())) {
            map.put(TYPE, IotdbDataType.DOUBLE.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.FLOAT.getType())) {
            map.put(TYPE, IotdbDataType.FLOAT.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.INTEGER.getType())) {
            map.put(TYPE, IotdbDataType.INT32.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.LONG.getType())
              || type.equalsIgnoreCase(IotdbDataType.INT64.getDescp())) {
            map.put(TYPE, IotdbDataType.INT64.getDescp());
          } else if (type.equalsIgnoreCase(MeasureParaTypeEnum.SHORT.getType())) {
            map.put(TYPE, IotdbDataType.INT32.getDescp());
          }
        } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
          // 如果是保留表，类型直接用表中的字段类型，不需要转换
          map.put(TYPE, columnMappingDto.getDstColumnType());
        }
        list.add(map);
      }
      dataxIotdbParasPojo.setColumn(list);
      writerColumns.add(dataxIotdbParasPojo);
    }
    dataxParam.setWriterColumns(writerColumns);

    dataxParam.setBatchSize(batchSize == null ? 50 : batchSize);
    dataxParam.setSourceType(JdbcConstants.KAFKA);
    return dataxParam;
  }
}
